package com.atcyj.gmall.realtime.dwd.db.app;

import com.atcyj.gamll.realtime.common.base.BaseSqlApp;
import com.atcyj.gamll.realtime.common.constant.Constant;
import com.atcyj.gamll.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author: cyj
 * @date: 2024/8/17
 */
public class DwdTradeOrderDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(
                10014,
                4,
                "dwd_trade_order_detail"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {

        // 设置状态的存活时间
        // ttl设置为多少合适？？ 这四张表的数据是提交下单业务过程中生成的（在同一个事务内同时生成的），肯定能关联上，但是需要考虑到网络延迟等问题，适当设置一下ttl即可，设置为5s
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5L));
        // 1. 创建动态表，读取topic_db主题数据
        createTopicDB(groupId, tableEnv);

        // 2. 从topic_db表中过滤出订单明细表数据
        Table orderDetailTable = tableEnv.sqlQuery(
                "select\n" +
                        "    `data`['id'] id,\n" +
                        "    `data`['order_id'] order_id,\n" +
                        "    `data`['sku_id'] sku_id,\n" +
                        "    `data`['sku_name'] sku_name,\n" +
                        "     cast(cast(`data`['order_price'] as decimal(16,2)) * cast( `data`['sku_num'] as decimal(16,2)) as string) split_original_amount,\n" +
                        "    `data`['sku_num'] sku_num,\n" +
                        "    `data`['create_time'] create_time,\n" +
                        "    `data`['split_total_amount'] split_total_amount,\n" +
                        "    `data`['split_activity_amount'] split_activity_amount,\n" +
                        "    `data`['split_coupon_amount'] split_coupon_amount,\n" +
                        "     ts \n" +
                        "from topic_db\n" +
                        "where `database`='gmall'\n" +
                        "  and `table`= 'order_detail'\n" +
                        "  and `type`='insert' ");
        tableEnv.createTemporaryView("order_detail", orderDetailTable);

        //  3. 从topic_bd表中过滤出订单表数据
        Table orderInfoTable = tableEnv.sqlQuery(
                "select\n" +
                        "  `data`['id'] id,\n" +
                        "  `data`['user_id'] user_id,\n" +
                        "  `data`['province_id'] province_id\n" +
                        "from topic_db\n" +
                        "where `database`='gmall'\n" +
                        "  and `table`= 'order_info'\n" +
                        "  and `type`='insert' ");
        tableEnv.createTemporaryView("order_info", orderInfoTable);

        // 4. 从topic_db表中过滤出订单明细活动关联表
        Table orderDetailActivityTable = tableEnv.sqlQuery(
                "select\n" +
                        "  `data`['order_detail_id'] order_detail_id,\n" +
                        "  `data`['activity_id'] activity_id,\n" +
                        "  `data`['activity_rule_id'] activity_rule_id\n" +
                        "from topic_db\n" +
                        "where `database`='gmall'\n" +
                        "  and `table`= 'order_detail_activity'\n" +
                        "  and `type`='insert'   ");
        tableEnv.createTemporaryView("order_detail_activity", orderDetailActivityTable);
        // 5. 从topic_db表中过滤出订单明细优惠券关联表
        Table orderDetailCouponTable = tableEnv.sqlQuery(
                "select\n" +
                        "  `data`['order_detail_id'] order_detail_id,\n" +
                        "  `data`['coupon_id'] coupon_id\n" +
                        "from topic_db\n" +
                        "where `database`='gmall'\n" +
                        "  and `table`= 'order_detail_coupon'\n" +
                        "  and `type`='insert'   \n");
        tableEnv.createTemporaryView("order_detail_coupon", orderDetailCouponTable);
        // 6. 关联这4张表的数据，用regular join
        Table result = tableEnv.sqlQuery(
                "select\n" +
                        "     od.id,\n" +
                        "     od.order_id,\n" +
                        "     oi.user_id,\n" +
                        "     oi.province_id,\n" +
                        "     od.sku_id,\n" +
                        "     od.sku_name,\n" +
                        "     od.split_original_amount,\n" +
                        "     od.sku_num,\n" +
                        "     oda.activity_id,\n" +
                        "     oda.activity_rule_id,\n" +
                        "     odc.coupon_id,\n" +
                        "     od.create_time,\n" +
                        "     od.split_total_amount,\n" +
                        "     od.split_activity_amount,\n" +
                        "     od.split_coupon_amount,\n" +
                        "     od.ts\n" +
                        "from order_detail od\n" +
                        "inner join order_info oi on od.order_id = oi.id \n" +
                        "left join  order_detail_activity oda on od.id = oda.order_detail_id\n" +
                        "left join  order_detail_coupon  odc on od.id = odc.order_detail_id");

        // 7. 创建kafka输出对应的动态表
        tableEnv.executeSql(
                "create table " + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + "(\n" +
                        "      id string ,\n" +
                        "      order_id string ,\n" +
                        "      user_id string ,\n" +
                        "      province_id string ,\n" +
                        "      sku_id string ,\n" +
                        "      sku_name string ,\n" +
                        "      split_original_amount string ,\n" +
                        "      sku_num string ,\n" +
                        "      activity_id string ,\n" +
                        "      activity_rule_id string ,\n" +
                        "      coupon_id string ,\n" +
                        "      create_time string ,\n" +
                        "      split_total_amount string ,\n" +
                        "      split_activity_amount string ,\n" +
                        "      split_coupon_amount string ,\n" +
                        "      ts bigint, \n" +
                        "      primary key(id) not enforced " +   // 定义主键
                        ") " + SQLUtil.getKafkaUpsertSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
//        "WITH (\n" +
//                "  'connector' = 'upsert-kafka',\n" +
//                "  'topic' = '" + topicName + "',\n" +
//                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
//                "  'key.format' = 'json',\n" +
//                "  'value.format' = 'json'\n" +
//                ")";


        // 8. 将查询结果写入到动态表中，即写出到kafka主题中
        result.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL).execute();
    }
}
